package com.gitee.ixtf.rsocket.verticle

import com.gitee.ixtf.core.Jassert
import com.gitee.ixtf.rsocket.BROKER_PORT
import com.gitee.ixtf.rsocket.BrokerAuth
import com.gitee.ixtf.rsocket.broker.*
import com.gitee.ixtf.vertx.verticle.BaseCoroutineVerticle
import io.rsocket.*
import io.rsocket.core.RSocketServer
import io.rsocket.frame.decoder.PayloadDecoder
import io.rsocket.transport.ServerTransport
import io.rsocket.transport.netty.server.TcpServerTransport
import io.vertx.ext.auth.jwt.JWTAuth
import jakarta.inject.Inject
import kotlinx.coroutines.reactor.awaitSingle
import kotlinx.coroutines.reactor.mono
import reactor.core.publisher.Mono

abstract class RSocketServerVerticle : BaseCoroutineVerticle(), SocketAcceptor, CloudEventHandler {
  private lateinit var closeable: Closeable
  @field:BrokerAuth @Inject protected open lateinit var authProvider: JWTAuth

  protected open fun buildServerTransport(): ServerTransport<out Closeable> =
      TcpServerTransport.create(BROKER_PORT)

  override suspend fun start() {
    super.start()
    println("===== ${this::class.java.simpleName} =====")
    closeable =
        RSocketServer.create(this)
            .payloadDecoder(PayloadDecoder.ZERO_COPY)
            .bind(buildServerTransport())
            .awaitSingle()
  }

  override suspend fun stop() {
    closeable.dispose()
    super.stop()
  }

  override fun accept(setup: ConnectionSetupPayload, sendingSocket: RSocket): Mono<RSocket> = mono {
    val cloudEvent = SetupRequest.decode(setup)
    val principal = SetupRequest.principal(cloudEvent, authProvider)
    Jassert.nonNull(principal)
    this@RSocketServerVerticle
  }
}
